package org.hadoop.input.excel;
import java.io.IOException;
import java.io.InputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.poi.hssf.usermodel.HSSFCell;
import org.apache.poi.hssf.usermodel.HSSFRow;
import org.apache.poi.hssf.usermodel.HSSFSheet;
import org.apache.poi.hssf.usermodel.HSSFWorkbook;
import org.apache.poi.ss.usermodel.CellType;
/**
 * 读取Excel数据的InputFormat<br>
 * 读取Excel中的所有Sheet然后再读取每一个Sheet中的所有行和列将数据转存成文本数据
 */
public class ExcelInputFormat extends FileInputFormat<LongWritable, Text> {
    @Override
    public RecordReader<LongWritable, Text> createRecordReader(InputSplit split, TaskAttemptContext context)
            throws IOException, InterruptedException {
        RecordReader<LongWritable, Text> rr = new ExcelRecordReader();//不需要调用initialize方法
        return rr;
    }
    //RecorderReader类的子类
    public static class ExcelRecordReader extends RecordReader<LongWritable, Text> {
        private HSSFWorkbook book;//表示Excel文件
        private int sheets = -1;//sheet的个数
        private int currentSheetIndex = 0;//已经读取到当前Sheet的下标
        private HSSFSheet sheet;//当前Sheet
        private int lastRowNum = -1;//每一个Sheet最后的行号
        private int currentRow = -1;//当前Sheet的当前行号
        private Text value = new Text();//这一行的数据
        @Override
        public void initialize(InputSplit split, TaskAttemptContext context)
                throws IOException, InterruptedException {
            FileSplit fileSplit = (FileSplit) split;
            FileSystem fs = FileSystem.get(context.getConfiguration());
            InputStream in = fs.open(fileSplit.getPath());
            book = new HSSFWorkbook(in);
            sheets = book.getNumberOfSheets();// 获取Sheet的个数
            if (sheets >= 1) {
                sheet = book.getSheetAt(currentSheetIndex++);// 获取第一个Sheet
                lastRowNum = sheet.getLastRowNum();// 最后一行的行号
                if (lastRowNum >= 0) {// 如果有数据则将当前行先设置为0
                    currentRow = 0;
                }
            }
        }
        @Override
        public boolean nextKeyValue() throws IOException, InterruptedException {
            if (sheet != null && currentRow <= lastRowNum) {
                HSSFRow row = sheet.getRow(currentRow++);
                short cells = row.getLastCellNum();
                value.set("");
                for (int i = 0; i < cells; i++) {
                    HSSFCell cell = row.getCell(i);
                    String str = "";
                    if (cell.getCellType() == CellType.STRING) {
                        str = cell.getStringCellValue();
                    } else {
                        str = "" + cell.getNumericCellValue();
                    }
                    value.set(value.toString() + "|" + str);
                }
                return true;
            } else {// 下一个Sheet
                if (currentSheetIndex < sheets) {
                    sheet = book.getSheetAt(currentSheetIndex++);// 获取第一个Sheet
                    if (sheet == null) {
                        return false;
                    }
                    lastRowNum = sheet.getLastRowNum();// 最后一行的行号
                    if (lastRowNum >=0) {// 如果有数据则将当前行先设置为0
                        currentRow = 0;
                    }
                    // 数据
                    if (currentRow <= lastRowNum) {// 有数据
                        HSSFRow row = sheet.getRow(currentRow++);
                        if(row==null) {//已经没有数据了
                            return false;
                        }
                        short cells = row.getLastCellNum();
                        value.set("");
                        for (int i = 0; i < cells; i++) {
                            HSSFCell cell = row.getCell(i);
                            String str = "";
                            if (cell.getCellType() == CellType.STRING) {
                                str = cell.getStringCellValue();
                            } else {
                                str = "" + cell.getNumericCellValue();
                            }
                            value.set(value.toString() + "|" + str);
                        }
                        return true;
                    }
                }
                return false;
            }
        }
        @Override
        public LongWritable getCurrentKey() throws IOException, InterruptedException {
            return new LongWritable(currentRow);
        }
        @Override
        public Text getCurrentValue() throws IOException, InterruptedException {
            return value;
        }
        @Override
        public float getProgress() throws IOException, InterruptedException {
            return 0;
        }
        @Override
        public void close() throws IOException {
            if (book != null) {
                book.close();
            }
        }
    }
}